package com.myself.consumer;

import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.common.serialization.StringDeserializer;

import java.util.Collections;
import java.util.Properties;

/**
 * @author W.S.
 * @version v1.0.0
 * @description 简单的消费端demo
 * @date Created in 2019/4/23 15:01
 */
public class ConsumerDemo {
    public static void main(String[] args) {
        Properties config = new Properties();
        config.setProperty("bootstrap.servers","127.0.0.1:9092");
        // key 反序列化
        config.setProperty("key.deserializer", StringDeserializer.class.getName());
        // value 反序列化
        config.setProperty("value.deserializer", StringDeserializer.class.getName());
        // 定义消费者群组
        config.setProperty("group.id", "1111");

        KafkaConsumer<String, String> consumer = new KafkaConsumer<String, String>(config);
        // 声明订阅那个主题
        consumer.subscribe(Collections.singletonList("my-topic"));
        while (true) {
            // 每次间隔多久拉取一次消息
            ConsumerRecords<String,String> records = consumer.poll(500);
            for (ConsumerRecord<String,String> record : records){
                System.out.println("$$$$$$$$$$$$$$$$$$$$$$$$4");
                System.out.println("消息所在主题："+record.topic());
                System.out.println("消息所在分区："+record.partition());
                System.out.println("消息key："+record.key());
                System.out.println("消息value："+record.value());
                System.out.println("消息偏移量："+record.offset());
            }
        }

    }
}
